//import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
//import org.apache.kafka.common.serialization.StringDeserializer
//import org.apache.spark.SparkConf
//import org.apache.spark.streaming.kafka010.KafkaUtils
//import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
//import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
//import org.apache.spark.streaming.{Seconds, StreamingContext}
//import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
//
//import java.util
//
//object HelloSparkStream {
//
//
//  def main(args: Array[String]): Unit = {
//
//    /*  maven 依赖  之前 spark-core
//        spark streaming   spark-streaming jar
//        spark-streaming-kafka
//
//       1) 创建 sparkconf 配置项 本地，应用程序名字
//       2） 创建 sparkStreaming  加载conf （ 实时的数据流）  ssc
//       3)  设置 编码  日志级别 error
//       4）  设置 kafka topic ，key value 反序列化， groupid ，offset 提交
//       5） 订阅 kafka的 消费者  topic ssc
//       6） 消费topic 当中的数据
//
//
//     */
//    //创建 sparkconf 配置项 本地，应用程序名字
//    val conf = new SparkConf().setMaster("local[*]").setAppName("helloStream")
//    //创建 sparkStreaming  加载conf （ 实时的数据流）  ssc
//    val ssc = new StreamingContext(conf,Seconds(2))
//
//    // error
//    ssc.sparkContext.setLogLevel("error")
//
//    // kafka 配置
//    val kfkParams = Map[String,Object](
//      "bootstrap.servers" ->"192.168.226.128:9092",
//      "key.deserializer"  -> classOf[StringDeserializer],
//      "value.deserializer" -> classOf[StringDeserializer],
//      "group.id" -> "niit",
//      "enable.auto.commit" -> (false:java.lang.Boolean)
//    )
//
//    // topic
//    val topicName = Array("16homework4")
//    //    val topicName = Array("test1")
//
//    // kafkaUtils  导入一个依赖 spark-streaming-kafka
//    val streamRdd = KafkaUtils.createDirectStream[String,String](
//      ssc,
//      PreferConsistent, // 位置策略
//      Subscribe[String,String](topicName,kfkParams)  // 订阅topic的名字
//    )
//
//
//    //streamRdd  kafka 返回的对象，值只是在value    RDD（key,value ）
//    //
//
//
//
//    //    val line =streamRdd.foreachRDD(
//    //      x=>{
//    //        if(x.isEmpty()){
//    //          val rs1 = x.map(_.value())
//    //        }
//    //      }
//    //    )
//    //时间窗口函数
//    val line = streamRdd.map(_.value())
//
//    val res = line.flatMap(_.split(" ")).map((_,1)).reduceByKeyAndWindow(
//      _+_,
//      Seconds(4),
//      Seconds(4)
//    )
//
//    //  producer 配置项
//    val property = new util.HashMap[String,Object]()
//    property.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.192.138:9092")
//    property.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
//    property.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
//
//
//
//    //res
//    res.foreachRDD(
//      x=>{
//        println("---------数据结果---------")
//        x.foreach(println(_))
//
//      }
//    )
//    ssc.start()  //开启 ssc
//    ssc.awaitTermination() //等待数据输入
//    line
//
//
//
//    //    streamRdd.foreachRDD(
//    //      x=>{
//    //        // 不为空的时候 进行打印
//    //        if(!x.isEmpty()){
//    //          // 取 kafka当中的 value ，为实际的数据
//    //          val line = x.map(_.value())
//    //          // 对一行数据进行打印
//    //          //          line.foreach(println)
//    //          // hello java
//    //          // 词频统计
//    //          val result = line.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)
//    //          result.foreach(println)
//    //
//    //
//    //
//    //          result.foreach(
//    //            obj =>{
//    //              // obj  元素
//    //
//    //              val kfkProducer = new KafkaProducer[String,String](property)
//    //              kfkProducer.send(   new ProducerRecord[String,String]    (  "test1",obj.toString()   ))
//    //              kfkProducer.close()
//    //
//    //
//    //            }
//    //          )
//    //
//    //
//    //
//    //        }
//    //      }
//    //    )
//    //
//    //
//    //     ssc.start()  //开启 ssc
//    //     ssc.awaitTermination() //等待数据输入
//
//
//  }
//
//}
